mod proto;

use tonic::transport::{Endpoint, Channel};
use proto::hello::{say_client::SayClient,HelloRequest};
use tonic::{Request, Status};
use tower::discover::Change;
use tonic::metadata::AsciiMetadataValue;
use tokio::sync::mpsc::Sender;
use std::sync::atomic::{AtomicBool, Ordering};
use prost::alloc::sync::Arc;

#[tokio::main]
async fn main()-> Result<(), Box<dyn std::error::Error>> {

    let (channel,sender) = Channel::balance_channel(10);
    let status = Arc::new(AtomicBool::new(false));

    let mut client = SayClient::with_interceptor(channel, add_request_id(add_token));
    start_channel_change_task(sender,status.clone());
    start_request_task(client,status).await;
    Ok(())
}

fn start_channel_change_task(sender:Sender<Change<i32,Endpoint>>,status:Arc<AtomicBool>){
    tokio::spawn(async move{
        // tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        wd_log::log_debug_ln!("-------------->添加6661端口服务");
        let ep = new_endpoint("http://[::1]:6661", 10, 3).unwrap();
        sender.send(Change::Insert(1,ep)).await;

        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        wd_log::log_debug_ln!("-------------->添加6662端口服务");
        let ep = new_endpoint("http://[::1]:6662", 10, 3).unwrap();
        sender.send(Change::Insert(2,ep)).await;

        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        wd_log::log_debug_ln!("==============>移除6661端口服务");
        sender.send(Change::Remove(1)).await;

        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        wd_log::log_debug_ln!("==============>移除6662端口服务");
        sender.send(Change::Remove(2)).await;

        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        wd_log::log_debug_ln!("-------------->添加6661端口服务");
        let ep = new_endpoint("http://[::1]:6661", 10, 3).unwrap();
        sender.send(Change::Insert(3,ep)).await;

        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        wd_log::log_debug_ln!("-------------->添加6662端口服务");
        let ep = new_endpoint("http://[::1]:6662", 10, 3).unwrap();
        sender.send(Change::Insert(1,ep)).await;

        tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        status.swap(true,Ordering::Relaxed)
    });
}

async fn start_request_task(mut client: SayClient<Channel>, status:Arc<AtomicBool>){
    while !status.load(Ordering::Relaxed) {
        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        let request = Request::new(HelloRequest{value:"hello".to_string()});
        match client.say_hello(request).await{
            Ok(o)=>{
                wd_log::log_info_ln!("response:{}",o.get_ref().value);
            }
            Err(e)=>{
                wd_log::log_error_ln!("response error:{}",e.to_string());
            }
        }
    }
}

//设置连接参数
fn new_endpoint(addr:&'static str,limit:usize,time_out:u64)->Result<Endpoint, Box<dyn std::error::Error>> {
    let ep = Endpoint::new(addr).unwrap()
        .concurrency_limit(limit)
        .timeout(std::time::Duration::from_secs(time_out));
    Ok(ep)
}
//添加token拦截器
fn  add_token(mut reqs:Request<()>) -> Result<Request<()>, Status>{
    let token = AsciiMetadataValue::from_str("fTSCcnnfIAUiEaNVmgXro7bwKtz2DXsUwi6Dcvt74qc=").unwrap();
    reqs.metadata_mut().insert("authorization",token);
    Ok(reqs)
}

//添加request_id拦截器
fn add_request_id<F>(f: F) -> impl Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static
    where F: Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static
{
    return move |mut reqs:Request<()>|{
        let token = AsciiMetadataValue::from_str("2021061201001").unwrap();
        reqs.metadata_mut().insert("request_id",token);
        f(reqs)
    }
}